package com.lagou.scala.test.Q5

import akka.actor.{Actor, PoisonPill}
case class Register(key: String)
case class Beat(key: String)
case object Ack
class Master extends Actor{
  val overTime = 5 //超时时间
  //存放worker信息map
  val workers = scala.collection.mutable.HashMap[String, Long]()
  //重写receive方法
  override def receive: Receive = {
    case Register(key) =>
      if (workers.contains(key)) {
        println(s"worker[$key] register conflict!")
      } else {
        updateHeartTime(key)
        println(s"worker[$key] 注册成功.")
        sender() ! Ack
      }
    case Beat(key) =>
      if (workers.contains(key)) {
        updateHeartTime(key)
        sender() ! Ack
      } else {
        sender() ! PoisonPill
      }
  }
  /**
   * 更新心跳时间
   */
  private def updateHeartTime(key: String) = {
    this.synchronized {
      workers(key) = System.currentTimeMillis()
    }
  }

  /**
   * 定时检查心跳时间，将超时的从 hashmap 中删除
   */
  private def checkTimestamp(): Unit = {
    this.synchronized {
      val now = System.currentTimeMillis()
      println("检查心跳：")
      for ((worker, ts) <- workers.toSet) {
        val duration = (now - ts) / 1000
        println(s"$worker => $duration 秒")
        if (duration > overTime) {
          println(s"$worker 已经超时")
          workers.remove(worker)
        }
      }
    }
  }
}
